package com.tandashi.springbootstudy01.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.BytesDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.utils.Bytes;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import java.util.Properties;

import static org.apache.kafka.clients.consumer.ConsumerConfig.*;

@Configuration
@Slf4j
public class KafkaConfig {

    @Bean
    public KafkaConsumer<String, Bytes> consumer(@Value("${kafka.bootstrap-servers}") String bootstrapServers){
        Properties props = new Properties();

        props.put(BOOTSTRAP_SERVERS_CONFIG,bootstrapServers);
        props.put(GROUP_ID_CONFIG,"CUSTOMER-SENSOR-CONSUMER-GROUP");
        props.put(ENABLE_AUTO_COMMIT_CONFIG,"false");
        props.put(AUTO_OFFSET_RESET_CONFIG,"earliest");
        props.put(MAX_POLL_INTERVAL_MS_CONFIG,"1800000");

        return new KafkaConsumer<>(props,new StringDeserializer(),new BytesDeserializer());
    }

}
